package io.kiki.sba.registry.client.provider;

import io.kiki.sba.registry.client.api.*;
import io.kiki.sba.registry.client.event.ConfiguratorProcessEvent;
import io.kiki.sba.registry.client.event.SubscriberProcessEvent;
import io.kiki.sba.registry.client.factory.NamedThreadFactory;
import io.kiki.sba.registry.client.task.ObserverHandler;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Setter
@Getter
public class DefaultObserverHandler implements ObserverHandler {


    private static final Logger logger = LoggerFactory.getLogger(DefaultObserverHandler.class);

    private static final int KEEP_ALIVE_TIME = 60;

    private ExecutorService executor;

    private EventBus eventBus;

    private ClientConfig clientConfig;


    public DefaultObserverHandler(ClientConfig clientConfig, EventBus eventBus) {
        this.clientConfig = clientConfig;
        this.executor = new ThreadPoolExecutor(clientConfig.getObserverThreadCoreSize(), clientConfig.getObserverThreadMaxSize(), KEEP_ALIVE_TIME, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>(clientConfig.getObserverThreadQueueLength()), new NamedThreadFactory("ObserverNotifyThread"));
        this.eventBus = eventBus;
    }


    @Override
    public void notify(Subscriber subscriber) {
        executor.submit(new SubscriberNotifyTask(subscriber));
    }


    @Override
    public void notify(Configurator configurator) {
        executor.submit(new ConfiguratorNotifyTask(configurator));
    }


    public class SubscriberNotifyTask implements Runnable {

        private Subscriber subscriber;


        public SubscriberNotifyTask(Subscriber subscriber) {
            this.subscriber = subscriber;
        }


        @Override
        public void run() {
            // ignore empty task
            if (null == subscriber) {
                return;
            }

            SubscriberProcessEvent event = new SubscriberProcessEvent();
            long start = System.currentTimeMillis();
            event.setStart(start);
            event.setClientConfig(clientConfig);
            event.setSubscriber(subscriber);
            try {
                SubscriberDataObserver dataObserver = subscriber.getSubscriberDataObserver();
                if (null != dataObserver) {
                    dataObserver.handleData(subscriber.getDataId(), subscriber.peekData());
                }
                logger.info("[notify] notify subscriber success, dataId: {}, registId:{}, cost: {}ms", subscriber.getDataId(), subscriber.getRegistId(), (System.currentTimeMillis() - start));
            } catch (Exception e) {
                logger.error("[notify] SubscriberNotifyTask execute error, dataId: {}", subscriber.getDataId(), e);
                event.setThrowable(e);
            } finally {
                event.setEnd(System.currentTimeMillis());
                if (null != eventBus) {
                    eventBus.post(event);
                }
            }
        }
    }


    public class ConfiguratorNotifyTask implements Runnable {

        private Configurator configurator;


        public ConfiguratorNotifyTask(Configurator configurator) {
            this.configurator = configurator;
        }

        @Override
        public void run() {
            if (null == configurator) {
                return;
            }

            ConfiguratorProcessEvent event = new ConfiguratorProcessEvent();
            long start = System.currentTimeMillis();
            event.setStart(start);
            event.setClientConfig(clientConfig);
            event.setConfigurator(configurator);
            try {
                ConfigDataObserver dataObserver = configurator.getDataObserver();
                if (null != dataObserver) {
                    dataObserver.handle(configurator.getDataId(), configurator.peekData());
                }

                logger.info("[notify] notify configurator success, dataId: {}, registId:{}, cost: {}ms", configurator.getDataId(), configurator.getRegistId(), (System.currentTimeMillis() - start));
            } catch (Exception e) {
                logger.error("[notify] ConfiguratorNotifyTask execute error, dataId: {}", configurator.getDataId(), e);
                event.setThrowable(e);
            } finally {
                event.setEnd(System.currentTimeMillis());
                if (null != eventBus) {
                    eventBus.post(event);
                }
            }
        }
    }
}
